package com.smsc.headend.task.engine.handler.impl.ansi;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.smsc.headend.common.enums.errorcode.TaskHandlerErrorCode;
import com.smsc.headend.common.utils.RedisKeys;
import com.smsc.headend.common.utils.RedisUtils;
import com.smsc.headend.module.asset.entity.Meter;
import com.smsc.headend.module.data.dto.AnsiDataDTO;
import com.smsc.headend.module.data.dto.DataItemDTO;
import com.smsc.headend.module.data.entity.RdMeterReadsLp;
import com.smsc.headend.module.data.enums.CollectionSource;
import com.smsc.headend.module.protocol.DeviceData;
import com.smsc.headend.module.protocol.ansi.AnsiData;
import com.smsc.headend.module.report.entity.LogScheduleTaskDevice;
import com.smsc.headend.module.task.dto.*;
import com.smsc.headend.module.task.enums.TaskType;
import com.smsc.headend.task.engine.dao.TaskDAO;
import com.smsc.headend.task.engine.exception.UDISTaskHandlerException;
import com.smsc.headend.task.engine.handler.TaskHandler;
import com.smsc.headend.task.engine.handler.impl.collection.strategy.CollectionStrategy;
import com.smsc.headend.task.engine.handler.impl.collection.strategy.CollectionStrategyFactory;
import com.smsc.headend.task.engine.service.DataProcessService;
import com.smsc.headend.task.engine.service.DeviceStatusService;
import com.smsc.headend.task.engine.service.FeignAssetManagementService;
import com.smsc.headend.task.engine.service.asset.AssetService;
import com.smsc.headend.task.engine.service.calc.RdEnergyDailyService;
import com.smsc.headend.task.engine.service.collection.CollectTaskService;
import com.smsc.headend.task.engine.service.task.LogScheduleTaskDeviceService;
import com.smsc.headend.task.engine.service.task.impl.TaskServiceImpl;
import com.smsc.headend.task.engine.utils.AnsiUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import java.util.function.Predicate;

import static com.smsc.headend.task.engine.handler.impl.collection.ReadLoadProfileBufferHandler.RECOLLECT_TASK_SEQ;

@Slf4j
public class ReadAnsiProfileDataHandler implements TaskHandler {
    @Autowired
    FeignAssetManagementService assetManagementService;
    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    DataProcessService dataProcessService;
    @Autowired
    TaskDAO taskDAO;
    @Autowired
    TaskServiceImpl taskService;
    @Autowired
    AssetService assetService;
    @Autowired
    LogScheduleTaskDeviceService logScheduleTaskDeviceService;
    @Autowired
    DeviceStatusService deviceStatusService;
    @Autowired
    RdEnergyDailyService rdEnergyDailyService;
    @Autowired
    RedisUtils redisUtils;
    @Autowired
    CollectionStrategyFactory collectionStrategyFactory;
    @Autowired
    CollectTaskService collectTaskService;

    @Override
    public int handle(Task task, AtomicTask atomicTask, DeviceData deviceData, String actionResult) throws UDISTaskHandlerException {
        if (!(deviceData instanceof AnsiData)) {
            log.error("result={}, data={}", actionResult, deviceData);
            throw new UDISTaskHandlerException(TaskHandlerErrorCode.DEVICE_DATA_TYPE_NOT_SUPPORT.toString());
        }
        AnsiData ansiData = (AnsiData) deviceData;
        AtomicTaskState atomicTaskState = taskDAO.hashGetAtomicTaskState(atomicTask.getAtomicTaskNo());
        Map<String, Object> param = getParameterMap(task, atomicTask);
        Long startIntervalTime = MapUtil.getLong(atomicTaskState.getData(), "START_INTERVAL_TIME");
        Integer intervalSize = MapUtil.getInt(atomicTaskState.getData(), "INTERVAL_SIZE");

        CollectionSource collectionSource = getCollectionType(task, param);
        Long lpTemplateId = MapUtil.getLong(param, "lpTemplateId");
        Long lpDataItemId = MapUtil.getLong(param, "dataItemId");
        List<DataItemDTO> dataItemAnsiList = dataProcessService.getLoadProfileCaptureObjects(lpTemplateId, lpDataItemId, task.getMeterId());
        Meter meter = assetService.getMeterById(task.getMeterId());
        String ansiTaskDataKey = RedisKeys.getAnsiTaskDataKey(task.getTaskNo());
        Map ansiTaskData = redisUtils.hmget(ansiTaskDataKey);

        Long profileStartTime = MapUtil.getLong(ansiTaskData, "profileStartTime");
        Long profileEndTime = MapUtil.getLong(ansiTaskData, "profileEndTime");
        String dataBufferKey = RedisKeys.getAnsiDataBufferKey(task.getTaskNo());
        int maxIntTimeSet1 = MapUtil.getInt(ansiTaskData, "MAX_INT_TIME_SET1");
        long interval = maxIntTimeSet1 * 60;
        List<Integer[]> ansiDataList = AnsiUtil.parseValue4ByteByIndex(ansiData.getAnsiByteBuffer(), intervalSize, dataItemAnsiList.size(), false);
        for (int i = 0; i < ansiDataList.size(); i++) {
            if (i > 0) {
                startIntervalTime += interval;
            }
            if (startIntervalTime < profileStartTime) {
                continue;
            }
            if(startIntervalTime >= profileEndTime) {
                break;
            }
            Integer[] list = ansiDataList.get(i);
            if(list[0] == -1) {
                continue;
            }
            redisUtils.lpush(dataBufferKey, JSONUtil.toJsonStr(new AnsiDataDTO(startIntervalTime, list)));
            log.info("push interval time: {}, parse data: {}", DateUtil.formatDateTime(new Date(startIntervalTime * 1000)), JSONUtil.toJsonStr(list));
        }
        redisUtils.expire(dataBufferKey, Long.valueOf(1 * 60 * 60));

        atomicTaskState.setCollectingEndTv(startIntervalTime);
        long nextIntervalTime = startIntervalTime + interval;
        if(nextIntervalTime < profileEndTime) {
            Integer blockSize = MapUtil.getInt(ansiTaskData, "BLOCK_SIZE");
            int currentIntervalOffset = MapUtil.getInt(atomicTaskState.getData(), "CURRENT_INTERVAL_OFFSET");
            if(currentIntervalOffset % blockSize == 0) {
                BlockInfo nextBlock = atomicTaskState.getMatchedBlockNumbers().poll();
                atomicTaskState.getData().put("CURRENT_BLOCK_NUM", nextBlock.getBlockNum());
                currentIntervalOffset = nextBlock.getOffset();
                nextIntervalTime = nextBlock.getFirstTime();
                atomicTaskState.getData().put("END_POINT_NUM", 0);
            }
            atomicTaskState.getData().put("CURRENT_INTERVAL_OFFSET", currentIntervalOffset);
            atomicTaskState.getData().put("START_INTERVAL_TIME", nextIntervalTime);
            taskDAO.hashSetAtomicTaskState(atomicTask.getAtomicTaskNo(), task, atomicTaskState.getRetryTimes(), atomicTaskState);
            return 0;
        } else {
            if (!atomicTaskState.getMatchedBlockNumbers().isEmpty()) {
                log.info("go to next block {}", atomicTaskState.getMatchedBlockNumbers().getFirst().getBlockNum());
                BlockInfo nextBlock = atomicTaskState.getMatchedBlockNumbers().poll();
                atomicTaskState.getData().put("CURRENT_BLOCK_NUM", nextBlock.getBlockNum());
                atomicTaskState.getData().put("CURRENT_INTERVAL_OFFSET", nextBlock.getOffset());
                atomicTaskState.getData().put("START_INTERVAL_TIME", nextIntervalTime);
                atomicTaskState.getData().put("END_POINT_NUM", 0);
                taskDAO.hashSetAtomicTaskState(atomicTask.getAtomicTaskNo(), task, atomicTaskState.getRetryTimes(), atomicTaskState);
                return 0;
            }
        }

        try {
            List<AnsiDataDTO> ansiDataDTOList = new ArrayList<>();
            Object data;
            while ((data = redisUtils.lpop(dataBufferKey)) != null) {
                AnsiDataDTO map = JSONUtil.toBean(String.valueOf(data), AnsiDataDTO.class);
                ansiDataDTOList.add(map);
            }

            // 去重
            List<AnsiDataDTO> distinctList = CollUtil.newArrayList();
            ansiDataDTOList.stream().filter(distinctByKey(p -> p.getIntervalTime())).forEach(distinctList::add);
            List<AnsiDataDTO> sortedList = CollUtil.sortByProperty(distinctList, "intervalTime");
            List<RdMeterReadsLp> meterReadsCol = dataProcessService.getAnsiRdMeterReadLpList(sortedList, lpTemplateId, lpDataItemId, dataItemAnsiList, meter, collectionSource);
            CollectionStrategy collectionStrategy = collectionStrategyFactory.getCollectionStrategy(collectionSource);
            //data filter
            List<RdMeterReadsLp> filteredRdMeterReadsLp = collectionStrategy.filterCollectionData(task, atomicTask, param, meterReadsCol);
            //save profile data
            TaskReadData readData = new TaskReadData();
            readData.setProfileId(lpDataItemId);
            readData.setTaskNo(task.getTaskNo());
            readData.setIsODR(task.getIsODR());
            if (CollUtil.isNotEmpty(filteredRdMeterReadsLp)) {
                deviceStatusService.updateMeterReadDataStatus(meter);
                readData.setTaskMeterReads(filteredRdMeterReadsLp);
                dataProcessService.saveTaskReadData(readData, collectionSource);
                if (TaskType.ScheduleRead_Profile.toString().equalsIgnoreCase(task.getTaskType())) {
                    logScheduleTaskDeviceService.sendMessage(task, LogScheduleTaskDevice.StatusEnum.Success.getCode(), LogScheduleTaskDevice.StatusDescEnum.CollectedSuccess.getCode());
                }
                rdEnergyDailyService.calcDailyEnergyInterval(readData);
            } else {
                if (TaskType.ScheduleRead_Profile.toString().equalsIgnoreCase(task.getTaskType())) {
                    logScheduleTaskDeviceService.sendMessage(task, LogScheduleTaskDevice.StatusEnum.Failed.getCode(), LogScheduleTaskDevice.StatusDescEnum.EmptyData.getCode());
                }
            }
            //callback
            int result = collectionStrategy.callback(meter, readData, task, atomicTask, param, dataItemAnsiList, filteredRdMeterReadsLp);
            return result;
        } catch (Exception e) {
            if (task.getIsODR()) {
                if (param.containsKey(RECOLLECT_TASK_SEQ)) {
                    Map<String, Object> data = MapUtil.newHashMap();
                    data.put("errorText", "Unexpected error");
                    taskDAO.addTaskData(task.getTaskNo(), atomicTask.getAtomicTaskNo(), param.get(RECOLLECT_TASK_SEQ).toString(), data, Long.valueOf(1 * 60 * 60), false);
                } else {
                    taskDAO.addTaskData(task.getTaskNo(), atomicTask.getAtomicTaskNo(), lpDataItemId.toString(), "Unexpected error", Long.valueOf(1 * 60 * 60), false);
                }
            }
            log.error("Unexpected error", e);

            return -1;
        }
    }

    static <T> Predicate<T> distinctByKey(Function<? super T, ?> keyExtractor) {
        Map<Object,Boolean> seen = new ConcurrentHashMap<>();
        //putIfAbsent方法添加键值对，如果map集合中没有该key对应的值，则直接添加，并返回null，如果已经存在对应的值，则依旧为原来的值。
        //如果返回null表示添加数据成功(不重复)，不重复(null==null :TRUE)
        return t -> seen.putIfAbsent(keyExtractor.apply(t), Boolean.TRUE) == null;
    }

    public static void main(String[] args) {
        long time = 1653442200;
        System.out.println((time - 900 * 20));
    }

    private CollectionSource getCollectionType(Task task, Map<String, Object> param) {
        if (TaskType.ScheduleRead_Profile.toString().equalsIgnoreCase(task.getTaskType())) {
            return CollectionSource.SCH;
        }

        if (TaskType.AutoReCollectProfile.toString().equalsIgnoreCase(task.getTaskType())) {
            return CollectionSource.ARR;
        }

        if (TaskType.ODR_Profile.toString().equalsIgnoreCase(task.getTaskType()) || TaskType.Interface_ODR_Profile.toString().equalsIgnoreCase(task.getTaskType())) {
            if (param.containsKey(RECOLLECT_TASK_SEQ)) {
                return CollectionSource.MARR;
            }
            return CollectionSource.ODR;
        }
        return null;
    }

    @Override
    public int handleFailure(Task task, AtomicTask atomicTask, String message) {
        log.info("error msg:{}", message);
        if (task.getIsODR()) {
            Map<String, Object> param = getParameterMap(task, atomicTask);
            Long lpDataItemId = Long.valueOf(String.valueOf(param.get("dataItemId")));
            // key 回曲线的dataItemId给web
            taskDAO.addTaskData(task.getTaskNo(), atomicTask.getAtomicTaskNo(), Convert.toStr(lpDataItemId), message, 3600l, false);
        }
        if (StrUtil.equalsIgnoreCase(task.getTaskType(), TaskType.ScheduleRead_Profile.name())) {
            logScheduleTaskDeviceService.sendMessage(task, LogScheduleTaskDevice.StatusEnum.Failed.getCode(), LogScheduleTaskDevice.StatusDescEnum.DecodeDataFailed.getCode());
        }
        return -1;
    }

}

